//
//  stats_container.cpp
// ~~~~~~~~~~~~~~~~~~~~~
//
// Copyright (c) 2013-2014 Razvan Coca (razvan dot coca at  gmail dot com)
// 


#include <boost/make_shared.hpp>
#include <sstream>
#include "logging/logging.hpp"
#include "stats_container.hpp"
#include "components/utils/proto_serializer.hpp"
namespace bt=boost::posix_time;

stats_container::stats_container(boost::asio::io_service& ios,std::string const & persist_path,bt::minutes expected_lifetime):
    timer_(ios),persist_path_(persist_path),expected_lifetime_(expected_lifetime)
{
    stats_map_ptr=boost::make_shared<stats_map_t>();
    load();
    boost::system::error_code ec;
    handle_timeout(ec);
}
void stats_container::handle_timeout(boost::system::error_code const&ec)
{
    if(!ec)
    {
        save();
        timer_.expires_from_now(expected_lifetime_);
        auto timed_save=boost::bind(&stats_container::handle_timeout,this,_1);
        timer_.async_wait(timed_save);
    }
}
bool stats_container::shutdown()
{
    timer_.cancel();
    return save();
}
/*
static std::string make_key(node_id const& node,std::string const& key)
{
    std::stringstream os;
    os<<node.cluster()<<"/"<<node.node()<<":"<<key;
    return std::string(os.str());
}*/

static std::string make_key(std::string const& key)
{
    return key;
}
std::string message_key(stats_message_t const & m)
{
    return make_key(m.data().name());
}

bool is_older_than(stats_message_t const & message,bt::minutes lifetime)
{
    boost::posix_time::ptime now(boost::posix_time::second_clock::universal_time());
    time_t timestamp=message.data().timestamp();
    boost::posix_time::ptime mtime=boost::posix_time::from_time_t(timestamp);
    return now-mtime>=lifetime;
}

bool stats_container::add(stats_message_t const& m)
{
    boost::mutex::scoped_lock lock(mutex_);
    auto insert_res=stats_map_ptr->insert(std::make_pair(message_key(m),m));
    //return insert_res.second;
    return true;
}

bool stats_container::get(std::string mkey,std::vector<stats_message_t>& m)
{
    boost::mutex::scoped_lock lock(mutex_);
    auto iter_pair=stats_map_ptr->equal_range(mkey);
    if(iter_pair.first==iter_pair.second) return false;

    for(auto i=iter_pair.first;i!=iter_pair.second;++i)
    {       
        m.push_back(i->second);
        m.back().set_type(stats_message_t::reply);
    }
    m.back().set_type(stats_message_t::end);

    return true;
}

int stats_container::purge_old()
{
    std::size_t old_items=0;
    auto new_map=boost::make_shared<stats_map_t>();
    boost::mutex::scoped_lock lock(mutex_);
    for(auto const& kv_pair:*stats_map_ptr)
    {
        if(!is_older_than(kv_pair.second,expected_lifetime_))
            new_map->insert(std::make_pair(make_key(kv_pair.second.data().name()),kv_pair.second));
        else ++old_items;
    }
    stats_map_ptr=new_map;
    return old_items;
}
bool stats_container::load()
{
    boost::mutex::scoped_lock lock(mutex_);
    {
        std::ifstream file(persist_path_,std::ios_base::binary);
        try{
            while(!file.eof())
            {
                auto item=load_item<stats_message_t>(file);
                stats_map_ptr->insert(std::make_pair(message_key(*item),*item));
            }

        }
        catch(std::exception const& e)
        {
            LOG(ERROR)<<"stats_container::load:"<<e.what();
            return false;
        }
        return true;
    }
}
bool stats_container::save()
{
    boost::mutex::scoped_lock lock(mutex_);
    std::ofstream file(persist_path_,std::ios_base::binary|std::ios_base::trunc);
    bool ret=true;
    for(auto const& kv_pair:*stats_map_ptr)
    {
        try{
            save_item(file,kv_pair.second);
        }
        catch(std::exception const& e)
        {
            LOG(ERROR)<<"stats_container::load:"<<e.what();
            ret=false;
        }
    }
    return ret;
}
